package com.foreveross.proxyip.core.pools;

import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;

import org.apache.commons.lang3.StringUtils;
import org.apache.poi.xslf.model.geom.SinArcTanExpression;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import redis.clients.jedis.Transaction;

import com.dayatang.domain.InstanceFactory;
import com.foreveross.proxyip.core.BaseCode;
import com.foreveross.proxyip.core.pools.validate.TestProxyIp;
import com.foreveross.proxyip.core.redis.IpStoreMap;
import com.foreveross.proxyip.core.redis.Map;
import com.foreveross.proxyip.core.redis.PiRedisExecption;
import com.foreveross.proxyip.core.redis.PoolMap;
import com.foreveross.proxyip.core.redis.RedisExecutor;
import com.foreveross.proxyip.core.redis.RedisManager;
import com.foreveross.proxyip.core.zk.ZkAppCenter;
import com.foreveross.proxyip.core.zk.task.strategy.CheckInTask;
import com.foreveross.proxyip.entity.ProxyIpBean;
import com.foreveross.proxyip.infra.base.ddd.repository.IBaseRepository;
import com.foreveross.proxyip.infra.conf.PropertyUtils;
import com.foreveross.proxyip.infra.conf.system.DefaultPropertiesLoader;

/**
 * 池定义接口
 * 
 * @author lyq
 * 
 */
@SuppressWarnings({ "rawtypes", "unchecked" , "unused"})
public abstract class AvailablePool {

	protected static final Logger log = LoggerFactory.getLogger(AvailablePool.class);
	
	private static final String UNIQUE_VERIFY_PRIFIX = "uniquenessVerifyPrifix_lock_";

	private static List<ProxyIpBean> pendingIps = new LinkedList<ProxyIpBean>();

	private static Boolean isOpenPendingIpThread = false;
	
	// public MemcachedManager memcached ;

	//public Thread checkInTh = null;
	protected List<Thread> checkInThs = Collections.synchronizedList(new ArrayList<Thread>());

	//public Thread checkOutTh = null;
	protected List<Thread> checkOutThs =  Collections.synchronizedList(new ArrayList<Thread>());

	protected Integer offerWaitTime = 0;
	
	protected int threadNum = 1;

	// 验证方式
	TestProxyIp validate = null;
	public String poolName = "";
	public String poolSign = "";

	public Integer checkInSepTime = 5000;
	
	public boolean openStatus = true;
	
	public boolean isOpenIpUniqueness = PropertyUtils.getBooleanProperty("proxyip.isOpenIpUniqueness", DefaultPropertiesLoader.CONFIG_FILE);
	public boolean isOpenChannelPoolCheckInVerify = PropertyUtils.getBooleanProperty("proxyip.isOpenChannelPoolCheckInVerify", DefaultPropertiesLoader.CONFIG_FILE);

	
	/*** 判断池是否正在执行checkIn，控制线程 ***/
	// public boolean checkIn = true,checkOut=false;

	protected Long proxyMax, proxyMin;
	public Long sourceNum = Long.MAX_VALUE;// 每个应用最多验证那么多ip;
	public Integer threadMaxIpCount;

	// public String sourceTask = "";

	// public String sourceCache = "";//redis源获取地址

	/******* ip实例集合，map类型保证了池中ip的唯一性 ********/
	// public final Map<String,ProxyIpBean> ipStoreMap = new
	// Hashtable<String,ProxyIpBean>();

	public Map poolIpMap = null;

	public IpStoreMap ipStoreMap = null;

	public RedisManager redisManager = RedisManager.getInstance();

	private Integer offerMaxQueue = 2000;
	
	//==================================================================
	//	   2014-09-01以后添加字段 	
	//==================================================================
	
	// 线程的检入与检出间隔时间
	public long checkInInterval;
	
	public long checkOutInterval;

	public AvailablePool(String poolName, String poolSign) {
		this.poolName = poolName;
		this.poolSign = poolSign.trim();
		ipStoreMap = IpStoreMap.getIntance();
		poolIpMap = new PoolMap(this.poolSign);
		log.info(poolName + "waiting.....");
	}

	/**
	 * 打开ip预处理线程的方法
	 */
	private void openPendingIpThread() {
		if (!isOpenPendingIpThread) {// 新增池是串行
			isOpenPendingIpThread = true;
			new Thread() {
				public void run() {
					while (true) {
						String ip = null;
						try {
							synchronized (pendingIps) {
								Integer size = pendingIps.size();
								if (size == 0) {
									pendingIps.wait();
								}
								ProxyIpBean proxyip = pendingIps.remove(0);
								ip = proxyip.getCompletionIP();
								if (isIpLock(ip)) {// 该ip正在被其它线程处理，滞后处理
									pendingIps.add(proxyip);
									continue;
								} else {
									update(proxyip);
								}
								size = pendingIps.size();
								if (size < offerMaxQueue && size > 0) {
									pendingIps.notifyAll();
								}
							}
						} catch (Exception e) {
							e.printStackTrace();
						} finally {
							try {
								unIpLock(ip);
							} catch (PiRedisExecption e) {
								e.printStackTrace();
							}
						}

					}

				}
			}.start();
		}
	}

	private void setPendingIp(final ProxyIpBean proxyIp) {
		synchronized (pendingIps) {
			Integer size = pendingIps.size();
			if (size >= offerMaxQueue) {
				try {
					pendingIps.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			pendingIps.add(proxyIp);//
			size = pendingIps.size();
			if (size < offerMaxQueue && size > 0) {
				pendingIps.notifyAll();
			}
		}
	}

	public abstract void initChannelProperty();

	public abstract void initBaseCode() throws NumberFormatException,
			PiRedisExecption;

	/**
	 * p集合大小
	 * 
	 * @return
	 */
	public String getPoolName() {
		return this.poolName;
	}

	public void destroy() {
		this.openStatus = false;
		try {
			// 遇到沉睡也会终止
			//this.checkInTh.interrupt();
			//this.checkOutTh.interrupt();
			// markfb
			for (Thread t : checkInThs) {
				t.interrupt();
			}
			
			for (Thread t : checkOutThs) {
				t.interrupt();
			}
		} catch (Exception e) {
			log.info(this.poolSign + "destory");
		}
		try {
			ZkAppCenter.getInstance().getZK()
					.delete(this.poolSign + "_CheckInTask", -1);
			ZkAppCenter.getInstance().getZK()
					.delete(this.poolSign + "_CheckOutTask", -1);
		} catch (Exception e) {
			log.debug(e.getMessage());
		}
		// markfb
		//this.checkInTh = null;
		//this.checkOutTh = null;
		this.checkInThs.clear();
		this.checkOutThs.clear();

	}

	public void open() {
		this.openStatus = true;
		checkOut();
		checkIn();
	}

	public abstract void checkIn();

	public abstract void checkOut();

	/**
	 * ip是否被锁住,一般锁住是正在被修改
	 * 
	 * @param ip
	 * @return
	 * @throws PiRedisExecption
	 */
	public Boolean isIpLock(String ip) throws PiRedisExecption {
		return this.redisManager.setnx(ip + "_lock", "LOCK", 10000) != 1;
	}

	public void unIpLock(String ip) throws PiRedisExecption {
		if (ip != null) {
			this.redisManager.del(ip + "_lock");
		}
	}

	/**
	 * 由池提供出一个IP
	 * 
	 * @param poolCode
	 * @param errCount
	 *            失败执行次数
	 * @param msg
	 * @return
	 * @throws Exception
	 */
	public ProxyIpBean offerIp(String poolCode, int errCount, StringBuffer msg)
			throws Exception {
		Random rand = new Random();
		Long beginTime = System.currentTimeMillis();
		
		while (size() == 0) {
			if (System.currentTimeMillis() - beginTime >= this.offerWaitTime) {// 获取ip超时
				log.error("{} [{}]获取ip超时{}" , new Object[] {this.poolName, this.poolSign ,this.offerWaitTime});
				if (msg != null) {
					msg.append("获取ip超时");
				}
				return null;
			}
		}
		
		List ip = null;
		// if((!PublicPool.POOL_CODE.equals(this.poolSign))&&!"default".equals(this.poolSign)){
		// ip = this.poolIpMap.subList(0L, 1L) ;//从拿第一个
		// }else{
		Long size = poolIpMap.size();
		size = size == null ? 0L : size;
		Integer i = rand.nextInt(size.intValue());
		ip = this.poolIpMap.subList(i.longValue(), i.longValue() + 1);// 从拿第一个;
		// }
		
		ProxyIpBean proxyIp = null;
		if (ip != null && ip.size() != 0) {
			proxyIp = (ProxyIpBean) ip.get(0);
		}
		
		if ((proxyIp == null || proxyIp.getRefuseChannel().contains(poolCode)) && errCount > 1) {
			// bean.getRefuseChannel().contains(null)==false
			// ip已经被拒绝过了
			return offerIp(poolCode, --errCount, msg);
		} else {
			// checkIpPoolRef(proxyIp);
			if (proxyIp == null) {
				log.info("offerIp获取ip为null");
				return null;
			}
			this.addProxyIpUserCount(proxyIp);
			if ((!PublicPool.POOL_CODE.equals(this.poolSign)) && !"default".equals(this.poolSign)) {
				// setPendingIp(proxyIp);
				try {
					while (System.currentTimeMillis() - beginTime <= 4000) {
						if (!this.isIpLock(proxyIp.getCompletionIP())) {
							break;
						}
					}
					//proxyIp = this.get(proxyIp.getCompletionIP());
					this.update(proxyIp);
				} finally {
					this.unIpLock(proxyIp.getCompletionIP());
				}
			}
			
			remove(proxyIp);
			
			return proxyIp;
		}
	}

	public void remove(final String ip) throws PiRedisExecption {
		new RedisExecutor<Long>() {
			@Override
			public Long run() throws Exception {
				tran = jedis.multi();
				poolIpMap.remove(tran, ip.trim());
				tran.exec();
				return null;
			}
		}.exec();
	}
	
	public void remove(final ProxyIpBean proxyIp) throws PiRedisExecption {
		if (!(this instanceof ChannelPool)) {
			return;
		} 
		
		final Integer maxUsedCount = Integer.parseInt( BaseCode.getValue("PUBLIC_MAX_USED_COUNT")); 
		new RedisExecutor<Long>() {
			@Override
			public Long run() throws Exception {
				tran = jedis.multi();
				poolIpMap.remove(tran, proxyIp.getCompletionIP());
				System.out.println(proxyIp.getChannelUseCount().get(PublicPool.POOL_CODE) + " --- " + maxUsedCount);
				if (proxyIp.getChannelUseCount().get(PublicPool.POOL_CODE) > maxUsedCount) {
					tran.del(proxyIp.getCompletionIP());
					System.out.println(proxyIp.getChannelUseCount().get(PublicPool.POOL_CODE) + " ------------- " + maxUsedCount);
				}
				
				tran.exec();
				return null;
			}
		}.exec();

		// Jedis jedis = null;
		// Transaction tran = null;
		// try{
		// jedis = redisManager.getJedis();
		//
		// }catch(Exception e){
		// tran.discard();
		// redisManager.getJedisPool().returnBrokenResource(jedis);
		// }finally{
		// redisManager.getJedisPool().returnResource(jedis);
		// }
	}

	/**
	 * 由池提供出一个IP
	 * 
	 * @return
	 * @throws Exception
	 */
	public ProxyIpBean offerIp(String poolCode) throws Exception {
		return this.offerIp(poolCode, 15, null);// 执行15次
	}

	/**
	 * 从池中提供num个ip
	 * 
	 * @param num
	 * @return
	 * @throws Exception
	 */
	public List offerIps(Long num, String poolCode) throws Exception {
		log.info("poolCode:'" + poolCode + "'get ips from '" + this.poolSign + "' begin.....");
		if (num <= 0) {
			return new ArrayList<ProxyIpBean>();
		}

		Long beginTime = System.currentTimeMillis();
		boolean bool = false;// 是否执行过等待加载ip
		
		while (size() <= num) {// 等待加载够ip
			bool = true;

			Long currTime = System.currentTimeMillis();
			if (currTime - beginTime >= this.offerWaitTime) {// 获取ip超时
				log.error("{} [{}]获取ip超时{}" , new Object[] {this.poolName, this.poolSign ,this.offerWaitTime});
				break;
			}
		}
		
		if (bool || size() <= num) {// 刚等待加载够数目的ip
			log.info("poolCode:'" + poolCode + "'get ips from '" + this.poolSign + "' end.....");
			return this.lastProcess((List) this.poolIpMap.subList(0L, num), poolCode);
		}
		List<ProxyIpBean> result = new ArrayList<ProxyIpBean>();
		return this.lastProcess(channelList(num.intValue(), poolCode, result, 1), poolCode);
	}

	/**
	 * 渠道需要获取的集合
	 * 
	 * @param num
	 * @param poolCode
	 * @param publicPoolList
	 * @return
	 * @throws Exception
	 */
	private List<ProxyIpBean> channelList(Integer num, String poolCode,
			List<ProxyIpBean> result, int count) throws Exception {
		if (num <= 0) {
			return new ArrayList<ProxyIpBean>();
		}
		Random rand = new Random();
		Integer startI = 0;
		Long currSize = size();
		Integer endSize = currSize.intValue() - num;
		Integer[] is = new Integer[endSize.intValue()];
		int j = 0;
		for (Integer i = num; i < currSize.intValue(); i++) {
			is[j] = i;
			j++;
		}
		Integer endI = is[rand.nextInt(is.length)];
		startI = endI - num;
		int remain = 0;
		result.addAll(this.subList(startI.longValue(), endI.longValue()));
		List<ProxyIpBean> tempList = new ArrayList<ProxyIpBean>(result);
		for (ProxyIpBean bean : tempList) {
			if (bean.getRefuseChannel().contains(poolCode)) {
				result.remove(bean);
				remain++;
			}
		}
		tempList.clear();
		if (remain > 0 && count < 2) {
			return this.channelList(remain, poolCode, result, ++count);
		} else {
			log.info("poolCode:'" + poolCode + "'get ips from '"
					+ this.poolSign + "' end.....");
			return result;
		}
	}


	private List<Integer> initCurrI(int size) {
		List<Integer> currI = new ArrayList<Integer>();
		for (int i = 0; i < size; i++) {
			currI.add(i);
		}
		return currI;
	}

	/**
	 * 提供结合时处理返回集合最后流程
	 * 
	 * @return
	 * @throws Exception
	 */
	protected List<ProxyIpBean> lastProcess(List<ProxyIpBean> returnList,
			String poolCode) throws Exception {
		for (ProxyIpBean proxyIp : returnList) {
			lastProcessAction(proxyIp, poolCode);
		}
		return returnList;
	}

	public abstract void lastProcessAction(ProxyIpBean bean, String poolCode)
			throws Exception;

	/**
	 * 获取池中的ip实例
	 * 
	 * @param ip
	 * @return
	 * @throws Exception
	 */
	public ProxyIpBean get(String ip) throws Exception {
		return (ProxyIpBean) ipStoreMap.get(ip);
	}
	
	

	// public void removeAction(ProxyIpBean bean){}

	// /**
	// * 列出池中的Ip集合
	// * @return
	// */
	// @SuppressWarnings({ "unchecked", "rawtypes" })
	// public List<ProxyIpBean> listAllIps(){
	// return new ArrayList( ipStoreMap.values());
	// }

	// /**
	// * 获取ip集合
	// * @return
	// */
	// public Set<String> getIpSet(){
	// return ipStoreMap.keySet();
	// }

	// public Map getIpStoreMap(){
	// return this.poolIpMap;
	// }
	public boolean contains(ProxyIpBean proxyIpBean) throws PiRedisExecption {
		return poolIpMap.containsKey(proxyIpBean.getCompletionIP());
	}

	public boolean contains(String ip) throws PiRedisExecption {
		return poolIpMap.containsKey(ip);
	}

	//
	// public boolean contains(Jedis jedis, String ip) {
	// return poolIpMap.containsKey(jedis,ip);
	// }

	/**
	 * ip集合大小
	 * 
	 * @return
	 * @throws PiRedisExecption
	 */
	public Long size() throws PiRedisExecption {
		return poolIpMap.size();
	}

	/**
	 * ip放进池里
	 * 
	 * @param ip
	 * @param proxyIpBean
	 * @throws Exception
	 */
	public void set(Transaction tran, ProxyIpBean proxyIpBean) throws Exception {
		proxyIpBean.getUsingKeys().add(this.poolSign);
		poolIpMap.set(tran, proxyIpBean.getCompletionIP());
	}
	
	public void set(Transaction tran, Integer score, ProxyIpBean proxyIpBean) throws Exception {
		proxyIpBean.getUsingKeys().add(this.poolSign);
		poolIpMap.set(tran, score == null ? 0 : score, proxyIpBean.getCompletionIP());
	}
	
	public void set(Transaction tran, String ip) throws Exception {
		poolIpMap.set(tran, ip);
	}

	public void set(ProxyIpBean proxyIpBean) throws PiRedisExecption {
		if (!poolIpMap.containsKey(proxyIpBean.getCompletionIP())) {// 已经存在了
			proxyIpBean.getUsingKeys().add(this.poolSign);
			poolIpMap.set(proxyIpBean.getCompletionIP());
		}
	}

	public void store(Transaction tran, ProxyIpBean proxyIpBean)
			throws Exception {
		this.ipStoreMap.set(tran, proxyIpBean.getCompletionIP(), proxyIpBean);
	}

	public Long store(ProxyIpBean proxyIpBean) throws PiRedisExecption {
		return this.ipStoreMap.set(proxyIpBean.getCompletionIP(), proxyIpBean);
	}

	public void remove(Transaction tran, ProxyIpBean bean) throws Exception {
		poolIpMap.remove(tran, bean.getCompletionIP());
	}

	// public void remove(ProxyIpBean bean) throws Exception{
	// Set<String> using = bean.getUsingKeys();
	// using.remove(this.poolSign);
	// if(using.isEmpty()&&this.ipStoreMap.containsKey(bean.getIp())){
	// this.ipStoreMap.remove(bean.getIp());
	// }
	// poolIpMap.remove(tran,bean.getIp());
	// }

	public int beforeUpdate(ProxyIpBean proxyIpBean) throws PiRedisExecption {
		return this.ipStoreMap.beforeUpdate(proxyIpBean.getCompletionIP(), proxyIpBean);
	}

	public String update(ProxyIpBean proxyIpBean) throws PiRedisExecption {
		return this.ipStoreMap.update(proxyIpBean.getCompletionIP(), proxyIpBean);
	}

	/**
	 * 装载iplist
	 * 
	 * @param ipList
	 * @throws Exception
	 */
	protected void setList(Transaction tran, List<ProxyIpBean> ipList)
			throws Exception {
		for (ProxyIpBean entity : ipList) {
			this.set(tran, entity);
		}
	}

	protected void setList(List<ProxyIpBean> ipList) throws PiRedisExecption {
		for (ProxyIpBean entity : ipList) {
			this.set(entity);
		}
	}

	public IBaseRepository getRepository() {
		IBaseRepository repository = InstanceFactory.getInstance(
				IBaseRepository.class, "entityHibernateRepository");
		return repository;
	}

	/**
	 * 获取需要checkIn的列表
	 * 
	 * @param num
	 * @return
	 * @throws Exception
	 */
	
	public List getCheckInList(Long num) throws PiRedisExecption {
		List list = new ArrayList();
		for (int i = 0; i < num; i++) {
			String ip = this.redisManager.rPop(CheckInTask.getSourceKey(this));
			if (StringUtils.isBlank(ip)) {
				continue;
			}
			
			list.add(ip);
		}
		
		return list;
	}

	public TestProxyIp getValidate() {
		return validate;
	}

	public void setValidate(TestProxyIp validate) {
		this.validate = validate;
	}

	/**
	 * 在proxy记录ip在该池使用次数
	 * 
	 * @param bean
	 */
	public void addProxyIpUserCount(ProxyIpBean bean) {
		if (PublicPool.POOL_CODE.equals(this.poolSign)) {// 公共池不需要
			return;
		}
		
		synchronized (AvailablePool.class) {
			Integer count = bean.getChannelUseCount().get(this.poolSign);
			if (count == null || count == 0) {
				bean.getChannelUseCount().put(this.poolSign, 1);
			} else {
				bean.getChannelUseCount().put(this.poolSign, count + 1);
			}
			Integer count2 = bean.getChannelUseCount().get(PublicPool.POOL_CODE);// 公共池也记录一下
			if (count2 == null || count2 == 0) {
				bean.getChannelUseCount().put(PublicPool.POOL_CODE, 1);
			} else {
				bean.getChannelUseCount().put(PublicPool.POOL_CODE, count2 + 1);
			}
		}

		// Integer publicCount =
		// bean.getChannelUseCount().get(PublicPool.POOL_CODE);
		// bean.getChannelUseCount().put(PublicPool.POOL_CODE, publicCount);
	}

	/**
	 * 记录ip第一次进入该池中时间
	 * 
	 * @param bean
	 */
	public void addIp2ChannelTime(ProxyIpBean bean) {
		Long intoTime = bean.getIp2ChannelTime().get(this.poolSign);
		if (intoTime == null || intoTime == 0) {
			bean.getIp2ChannelTime().put(this.poolSign,
					System.currentTimeMillis());
		}
	}

	/**
	 * 记录ip离开渠道池的时间
	 * 
	 * @param bean
	 * @throws PiRedisExecption
	 * @throws NumberFormatException
	 */
	public void addIpOutChannelTime(ProxyIpBean bean)
			throws NumberFormatException, PiRedisExecption {
		// Long outTime = bean.getIpOutChannelTime().get(this.poolSign);
		// if(outTime==null||outTime==0){
		if (!PublicPool.POOL_CODE.equals(this.poolSign)) {// 设置检出，冷却必须等待其过期
			this.redisManager.setnx(getIpOutTimeKey(bean),
					System.currentTimeMillis() + "", Integer.parseInt(BaseCode
							.getValue("CHANNEL_IP_OUT_SEP_TIME")) / 1000);
			// bean.getIpOutChannelTime().put(this.poolSign,System.currentTimeMillis());
		}
		// }
	}

	public String getIpOutTimeKey(ProxyIpBean bean) {
		return poolSign + "_" + bean.getCompletionIP() + "_" + "outChannel";
	}

	public String getIpLastCheckinTimeKey(ProxyIpBean bean) {
		return poolSign + "_" + bean.getCompletionIP() + "_" + "lastCheckin";
	}

	public List<ProxyIpBean> subList(Long start, Long end)
			throws PiRedisExecption {
		List<ProxyIpBean> list = this.poolIpMap.subList(start, end);
		return list;
	}

	public Long getProxyMax() throws Exception {
		return this.proxyMax; // * ZkAppCenter.getInstance().getAppsSize();
	}

	public Long getProxyMin() throws Exception {
		return proxyMin; // * ZkAppCenter.getInstance().getAppsSize();
	}

	
	//-------------------------------------------------
	// 说明:  2014-09-23后改版新添加方法
	
	/**
	 * 判断ip是否被己经分配到任意渠道当中
	 * 
	 * @param ip
	 * @return true : 己分配
	 * @throws Exception 
	 */
	public boolean isAlreadyDistributionChannel(String ip) throws Exception {
		int time = isOpenIpUniqueness ? 60000 * 60 * 2 : 6000;
		boolean result = true;
		long specifically = redisManager.setnx(UNIQUE_VERIFY_PRIFIX + ip, "LOCK", time);

		// 如果加锁成功, 表示现在还未有渠道分配使用
		if (specifically == 1L ) {
			result = false;
		}
		
		return result;
	}
	
	/**
	 * 删除此ip己被分配使用标记
	 * 
	 * @param ip
	 * @throws Exception
	 */
	public void removeMarkIpAlreadyUsed(String ip) throws Exception {
		redisManager.del(UNIQUE_VERIFY_PRIFIX + ip);
	}
	
	public boolean isAlreadyMarkUsed(String ip) throws PiRedisExecption {
		String result = redisManager.get(UNIQUE_VERIFY_PRIFIX + ip);
		
		return StringUtils.isNotBlank(result);
	}
}
